package com.king.first.app.pv;

import com.king.first.bean.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 网站总浏览量（PV）的统计
 */
public class Project_PV_WordCount {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> streamSource = env.readTextFile("input/UserBehavior.csv");

        SingleOutputStreamOperator<UserBehavior> filterStream = streamSource.map(value -> {
            String[] var = value.split(",");
            return new UserBehavior(Long.valueOf(var[0]), Long.valueOf(var[1]), Integer.valueOf(var[2]), var[3],
                    Long.valueOf(var[4]));
        }).filter(value -> "pv".equals(value.getBehavior()));

        SingleOutputStreamOperator<Tuple2<String, Long>> map = filterStream.map(value -> new Tuple2<>(value.getBehavior(), 1L)).returns(Types.TUPLE(Types.STRING, Types.LONG));
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = map.keyBy(value -> value.f0).sum(1);
        sum.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main1(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> streamSource = env.readTextFile("input/UserBehavior.csv");

        //894923,3076029,1879194,pv,1511658000
        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = streamSource.flatMap(
                (FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
                    String[] s1 = s.split(",");
                    if ("pv".equals(s1[3])) {
                        collector.collect(Tuple2.of(s1[3], 1));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT));
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = streamOperator.keyBy(key -> key.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> pv = keyedStream.sum(1);

        pv.print();


        env.execute();
    }
}
